/**
 * 
 */
package org.pump.client.main;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.future.CloseFuture;
import org.apache.mina.core.future.IoFuture;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.pump.client.codec.AuthPack;
import org.pump.client.codec.BinlogEvent;
import org.pump.client.codec.DumpBinlogCommand;
import org.pump.client.codec.EofPacket;
import org.pump.client.codec.ErrorPack;
import org.pump.client.codec.EventData;
import org.pump.client.codec.InitPack;
import org.pump.client.codec.OkPack;
import org.pump.client.codec.Packet;
import org.pump.client.codec.UnknownPack;
import org.pump.client.pool.PumpPool;
import org.pump.client.util.MyConstant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author liudong03
 * 
 */
public class ClientHandler extends IoHandlerAdapter {
	private static final Logger mylog = LoggerFactory
			.getLogger(ClientHandler.class);
	MyProperty conf;
	PumpPool eventPool;
	MyClient client;

	public boolean init(MyProperty conf, PumpPool eventPool, MyClient client) {
		this.conf = conf;
		this.eventPool = eventPool;
		this.client = client;
		return true;
	}

	public void sessionOpened(IoSession session) {
		mylog.debug(session + " opened");

	}

	public void messageReceived(IoSession session, Object message) {
		Packet mypack = (Packet) message;
		if (mypack.getBody() instanceof InitPack) {
			InitPack initpack = (InitPack) mypack.getBody();
			mylog.debug(initpack.toString());
			String randoms_seed = initpack.getSalt();

			AuthPack authpack = new AuthPack(this.conf, randoms_seed);
			mypack.getHead().setPacketlen(authpack.getLen());
			mypack.getHead().setPacketnum((byte) 0x01);
			mypack.setBody(authpack);
		} else if (mypack.getBody() instanceof OkPack) {
			OkPack respack = (OkPack) mypack.getBody();
			mylog.debug(respack.toString());

			DumpBinlogCommand dumpcmd = new DumpBinlogCommand(conf);
			mylog.debug(dumpcmd.toString());

			mypack.getHead().setPacketlen(dumpcmd.getLen());
			mypack.getHead().setPacketnum((byte) 0x00);
			mypack.setBody(dumpcmd);
		} else if (mypack.getBody() instanceof UnknownPack) {
			UnknownPack respack = (UnknownPack) mypack.getBody();
			mylog.info(respack.toString());
			mylog.error("unsupported packet");
			mypack = null;

		} else if (mypack.getBody() instanceof BinlogEvent) {
			BinlogEvent myevent = (BinlogEvent) mypack.getBody();
			this.eventPool.put(myevent);
			//mylog.info(mydata.toString());
			//session.close(false);
			mypack = null;
		} else if (mypack.getBody() instanceof ErrorPack) {
			ErrorPack errorpack = (ErrorPack) mypack.getBody();
			mylog.info(errorpack.toString());
			
		} else if (mypack.getBody() instanceof EofPacket) {
			EofPacket eofpack = (EofPacket) mypack.getBody();
			mylog.error("errno=[" + MyConstant.MASTER_STOP_TALK + "] message:receive eof packet,master stop communication,mypump stop work");
			System.exit(-1);
		} else {
			mylog.error("unsupported packet");
			mypack = null;
		}

		if (mypack != null) {
			session.write(mypack);
		}

	}

	public void exceptionCaught(IoSession session, Throwable cause) {
		mylog.error("exception" + session, cause);
	}
	
	public void sessionClosed(IoSession session) throws Exception {
		String close_type = (String) session.getAttribute(MyConstant.CLOSE_TYPE);
		if(close_type == null)
		{
			mylog.info("session passive colosed,we will reconnect..");
			if(reConnect(session))
			{
				return;
			}
		}
		
		this.client.connector.dispose();
    }
	public boolean reConnect(IoSession session)
	{
		for(int i = 0; i < MyConstant.repeat_count; i++)
		{
			try {
				Thread.sleep(MyConstant.SLEEP_INTERVAL);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				mylog.error("interrupted exception",e);
			}
			mylog.info("session reconnect [" + i + " times");
			if(this.client.start() == MyConstant.SUCC)
			{
				mylog.info("reconnect successfully");
				return true;
			}
		}
		return false;
	}
}
